/*
 * Copyright (c) 2013, 2014, Oracle and/or its affiliates. All rights reserved.
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */
package java.util.stream;

import java.util.DoubleSummaryStatistics;
import java.util.Objects;
import java.util.OptionalDouble;
import java.util.PrimitiveIterator;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.DoubleBinaryOperator;
import java.util.function.DoubleConsumer;
import java.util.function.DoubleFunction;
import java.util.function.DoublePredicate;
import java.util.function.DoubleToIntFunction;
import java.util.function.DoubleToLongFunction;
import java.util.function.DoubleUnaryOperator;
import java.util.function.IntFunction;
import java.util.function.ObjDoubleConsumer;
import java.util.function.Supplier;

/**
 * Abstract base class for an intermediate pipeline stage or pipeline source
 * stage implementing whose elements are of type {@code double}.
 *
 * @param <E_IN> type of elements in the upstream source
 * @since 1.8
 */
abstract class DoublePipeline<E_IN>
    extends AbstractPipeline<E_IN, Double, DoubleStream>
    implements DoubleStream {

  /**
   * Constructor for the head of a stream pipeline.
   *
   * @param source {@code Supplier<Spliterator>} describing the stream source
   * @param sourceFlags the source flags for the stream source, described in {@link StreamOpFlag}
   */
  DoublePipeline(Supplier<? extends Spliterator<Double>> source,
      int sourceFlags, boolean parallel) {
    super(source, sourceFlags, parallel);
  }

  /**
   * Constructor for the head of a stream pipeline.
   *
   * @param source {@code Spliterator} describing the stream source
   * @param sourceFlags the source flags for the stream source, described in {@link StreamOpFlag}
   */
  DoublePipeline(Spliterator<Double> source,
      int sourceFlags, boolean parallel) {
    super(source, sourceFlags, parallel);
  }

  /**
   * Constructor for appending an intermediate operation onto an existing
   * pipeline.
   *
   * @param upstream the upstream element source.
   * @param opFlags the operation flags
   */
  DoublePipeline(AbstractPipeline<?, E_IN, ?> upstream, int opFlags) {
    super(upstream, opFlags);
  }

  /**
   * Adapt a {@code Sink<Double> to a {@code DoubleConsumer}, ideally simply
   * by casting.
   */
  private static DoubleConsumer adapt(Sink<Double> sink) {
    if (sink instanceof DoubleConsumer) {
      return (DoubleConsumer) sink;
    } else {
      if (Tripwire.ENABLED) {
        Tripwire.trip(AbstractPipeline.class,
            "using DoubleStream.adapt(Sink<Double> s)");
      }
      return sink::accept;
    }
  }

  /**
   * Adapt a {@code Spliterator<Double>} to a {@code Spliterator.OfDouble}.
   *
   * @implNote The implementation attempts to cast to a Spliterator.OfDouble, and throws an
   * exception if this cast is not possible.
   */
  private static Spliterator.OfDouble adapt(Spliterator<Double> s) {
    if (s instanceof Spliterator.OfDouble) {
      return (Spliterator.OfDouble) s;
    } else {
      if (Tripwire.ENABLED) {
        Tripwire.trip(AbstractPipeline.class,
            "using DoubleStream.adapt(Spliterator<Double> s)");
      }
      throw new UnsupportedOperationException("DoubleStream.adapt(Spliterator<Double> s)");
    }
  }

  // Shape-specific methods

  @Override
  final StreamShape getOutputShape() {
    return StreamShape.DOUBLE_VALUE;
  }

  @Override
  final <P_IN> Node<Double> evaluateToNode(PipelineHelper<Double> helper,
      Spliterator<P_IN> spliterator,
      boolean flattenTree,
      IntFunction<Double[]> generator) {
    return Nodes.collectDouble(helper, spliterator, flattenTree);
  }

  @Override
  final <P_IN> Spliterator<Double> wrap(PipelineHelper<Double> ph,
      Supplier<Spliterator<P_IN>> supplier,
      boolean isParallel) {
    return new StreamSpliterators.DoubleWrappingSpliterator<>(ph, supplier, isParallel);
  }

  @Override
  @SuppressWarnings("unchecked")
  final Spliterator.OfDouble lazySpliterator(Supplier<? extends Spliterator<Double>> supplier) {
    return new StreamSpliterators.DelegatingSpliterator.OfDouble(
        (Supplier<Spliterator.OfDouble>) supplier);
  }

  @Override
  final void forEachWithCancel(Spliterator<Double> spliterator, Sink<Double> sink) {
    Spliterator.OfDouble spl = adapt(spliterator);
    DoubleConsumer adaptedSink = adapt(sink);
    do {
    } while (!sink.cancellationRequested() && spl.tryAdvance(adaptedSink));
  }

  @Override
  final Node.Builder<Double> makeNodeBuilder(long exactSizeIfKnown,
      IntFunction<Double[]> generator) {
    return Nodes.doubleBuilder(exactSizeIfKnown);
  }

  // DoubleStream

  @Override
  public final PrimitiveIterator.OfDouble iterator() {
    return Spliterators.iterator(spliterator());
  }

  @Override
  public final Spliterator.OfDouble spliterator() {
    return adapt(super.spliterator());
  }

  // Stateless intermediate ops from DoubleStream

  @Override
  public final Stream<Double> boxed() {
    return mapToObj(Double::valueOf);
  }

  @Override
  public final DoubleStream map(DoubleUnaryOperator mapper) {
    Objects.requireNonNull(mapper);
    return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
        StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
        return new Sink.ChainedDouble<Double>(sink) {
          @Override
          public void accept(double t) {
            downstream.accept(mapper.applyAsDouble(t));
          }
        };
      }
    };
  }

  @Override
  public final <U> Stream<U> mapToObj(DoubleFunction<? extends U> mapper) {
    Objects.requireNonNull(mapper);
    return new ReferencePipeline.StatelessOp<Double, U>(this, StreamShape.DOUBLE_VALUE,
        StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<U> sink) {
        return new Sink.ChainedDouble<U>(sink) {
          @Override
          public void accept(double t) {
            downstream.accept(mapper.apply(t));
          }
        };
      }
    };
  }

  @Override
  public final IntStream mapToInt(DoubleToIntFunction mapper) {
    Objects.requireNonNull(mapper);
    return new IntPipeline.StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
        StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Integer> sink) {
        return new Sink.ChainedDouble<Integer>(sink) {
          @Override
          public void accept(double t) {
            downstream.accept(mapper.applyAsInt(t));
          }
        };
      }
    };
  }

  @Override
  public final LongStream mapToLong(DoubleToLongFunction mapper) {
    Objects.requireNonNull(mapper);
    return new LongPipeline.StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
        StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Long> sink) {
        return new Sink.ChainedDouble<Long>(sink) {
          @Override
          public void accept(double t) {
            downstream.accept(mapper.applyAsLong(t));
          }
        };
      }
    };
  }

  @Override
  public final DoubleStream flatMap(DoubleFunction<? extends DoubleStream> mapper) {
    return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
        StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
        return new Sink.ChainedDouble<Double>(sink) {
          @Override
          public void begin(long size) {
            downstream.begin(-1);
          }

          @Override
          public void accept(double t) {
            try (DoubleStream result = mapper.apply(t)) {
              // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
              if (result != null) {
                result.sequential().forEach(i -> downstream.accept(i));
              }
            }
          }
        };
      }
    };
  }

  @Override
  public DoubleStream unordered() {
    if (!isOrdered()) {
      return this;
    }
    return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE, StreamOpFlag.NOT_ORDERED) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
        return sink;
      }
    };
  }

  @Override
  public final DoubleStream filter(DoublePredicate predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
        StreamOpFlag.NOT_SIZED) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
        return new Sink.ChainedDouble<Double>(sink) {
          @Override
          public void begin(long size) {
            downstream.begin(-1);
          }

          @Override
          public void accept(double t) {
            if (predicate.test(t)) {
              downstream.accept(t);
            }
          }
        };
      }
    };
  }

  @Override
  public final DoubleStream peek(DoubleConsumer action) {
    Objects.requireNonNull(action);
    return new StatelessOp<Double>(this, StreamShape.DOUBLE_VALUE,
        0) {
      @Override
      Sink<Double> opWrapSink(int flags, Sink<Double> sink) {
        return new Sink.ChainedDouble<Double>(sink) {
          @Override
          public void accept(double t) {
            action.accept(t);
            downstream.accept(t);
          }
        };
      }
    };
  }

  // Stateful intermediate ops from DoubleStream

  @Override
  public final DoubleStream limit(long maxSize) {
    if (maxSize < 0) {
      throw new IllegalArgumentException(Long.toString(maxSize));
    }
    return SliceOps.makeDouble(this, (long) 0, maxSize);
  }

  @Override
  public final DoubleStream skip(long n) {
    if (n < 0) {
      throw new IllegalArgumentException(Long.toString(n));
    }
    if (n == 0) {
      return this;
    } else {
      long limit = -1;
      return SliceOps.makeDouble(this, n, limit);
    }
  }

  @Override
  public final DoubleStream sorted() {
    return SortedOps.makeDouble(this);
  }

  @Override
  public final DoubleStream distinct() {
    // While functional and quick to implement, this approach is not very efficient.
    // An efficient version requires a double-specific map/set implementation.
    return boxed().distinct().mapToDouble(i -> (double) i);
  }

  // Terminal ops from DoubleStream

  @Override
  public void forEach(DoubleConsumer consumer) {
    evaluate(ForEachOps.makeDouble(consumer, false));
  }

  @Override
  public void forEachOrdered(DoubleConsumer consumer) {
    evaluate(ForEachOps.makeDouble(consumer, true));
  }

  @Override
  public final double sum() {
        /*
         * In the arrays allocated for the collect operation, index 0
         * holds the high-order bits of the running sum, index 1 holds
         * the low-order bits of the sum computed via compensated
         * summation, and index 2 holds the simple sum used to compute
         * the proper result if the stream contains infinite values of
         * the same sign.
         */
    double[] summation = collect(() -> new double[3],
        (ll, d) -> {
          Collectors.sumWithCompensation(ll, d);
          ll[2] += d;
        },
        (ll, rr) -> {
          Collectors.sumWithCompensation(ll, rr[0]);
          Collectors.sumWithCompensation(ll, rr[1]);
          ll[2] += rr[2];
        });

    return Collectors.computeFinalSum(summation);
  }

  @Override
  public final OptionalDouble min() {
    return reduce(Math::min);
  }

  @Override
  public final OptionalDouble max() {
    return reduce(Math::max);
  }

  /**
   * {@inheritDoc}
   *
   * @implNote The {@code double} format can represent all consecutive integers in the range
   * -2<sup>53</sup> to 2<sup>53</sup>. If the pipeline has more than 2<sup>53</sup> values, the
   * divisor in the average computation will saturate at 2<sup>53</sup>, leading to additional
   * numerical errors.
   */
  @Override
  public final OptionalDouble average() {
        /*
         * In the arrays allocated for the collect operation, index 0
         * holds the high-order bits of the running sum, index 1 holds
         * the low-order bits of the sum computed via compensated
         * summation, index 2 holds the number of values seen, index 3
         * holds the simple sum.
         */
    double[] avg = collect(() -> new double[4],
        (ll, d) -> {
          ll[2]++;
          Collectors.sumWithCompensation(ll, d);
          ll[3] += d;
        },
        (ll, rr) -> {
          Collectors.sumWithCompensation(ll, rr[0]);
          Collectors.sumWithCompensation(ll, rr[1]);
          ll[2] += rr[2];
          ll[3] += rr[3];
        });
    return avg[2] > 0
        ? OptionalDouble.of(Collectors.computeFinalSum(avg) / avg[2])
        : OptionalDouble.empty();
  }

  @Override
  public final long count() {
    return mapToLong(e -> 1L).sum();
  }

  @Override
  public final DoubleSummaryStatistics summaryStatistics() {
    return collect(DoubleSummaryStatistics::new, DoubleSummaryStatistics::accept,
        DoubleSummaryStatistics::combine);
  }

  @Override
  public final double reduce(double identity, DoubleBinaryOperator op) {
    return evaluate(ReduceOps.makeDouble(identity, op));
  }

  @Override
  public final OptionalDouble reduce(DoubleBinaryOperator op) {
    return evaluate(ReduceOps.makeDouble(op));
  }

  @Override
  public final <R> R collect(Supplier<R> supplier,
      ObjDoubleConsumer<R> accumulator,
      BiConsumer<R, R> combiner) {
    BinaryOperator<R> operator = (left, right) -> {
      combiner.accept(left, right);
      return left;
    };
    return evaluate(ReduceOps.makeDouble(supplier, accumulator, operator));
  }

  @Override
  public final boolean anyMatch(DoublePredicate predicate) {
    return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.ANY));
  }

  @Override
  public final boolean allMatch(DoublePredicate predicate) {
    return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.ALL));
  }

  @Override
  public final boolean noneMatch(DoublePredicate predicate) {
    return evaluate(MatchOps.makeDouble(predicate, MatchOps.MatchKind.NONE));
  }

  @Override
  public final OptionalDouble findFirst() {
    return evaluate(FindOps.makeDouble(true));
  }

  @Override
  public final OptionalDouble findAny() {
    return evaluate(FindOps.makeDouble(false));
  }

  @Override
  public final double[] toArray() {
    return Nodes.flattenDouble((Node.OfDouble) evaluateToArrayNode(Double[]::new))
        .asPrimitiveArray();
  }

  //

  /**
   * Source stage of a DoubleStream
   *
   * @param <E_IN> type of elements in the upstream source
   */
  static class Head<E_IN> extends DoublePipeline<E_IN> {

    /**
     * Constructor for the source stage of a DoubleStream.
     *
     * @param source {@code Supplier<Spliterator>} describing the stream source
     * @param sourceFlags the source flags for the stream source, described in {@link StreamOpFlag}
     * @param parallel {@code true} if the pipeline is parallel
     */
    Head(Supplier<? extends Spliterator<Double>> source,
        int sourceFlags, boolean parallel) {
      super(source, sourceFlags, parallel);
    }

    /**
     * Constructor for the source stage of a DoubleStream.
     *
     * @param source {@code Spliterator} describing the stream source
     * @param sourceFlags the source flags for the stream source, described in {@link StreamOpFlag}
     * @param parallel {@code true} if the pipeline is parallel
     */
    Head(Spliterator<Double> source,
        int sourceFlags, boolean parallel) {
      super(source, sourceFlags, parallel);
    }

    @Override
    final boolean opIsStateful() {
      throw new UnsupportedOperationException();
    }

    @Override
    final Sink<E_IN> opWrapSink(int flags, Sink<Double> sink) {
      throw new UnsupportedOperationException();
    }

    // Optimized sequential terminal operations for the head of the pipeline

    @Override
    public void forEach(DoubleConsumer consumer) {
      if (!isParallel()) {
        adapt(sourceStageSpliterator()).forEachRemaining(consumer);
      } else {
        super.forEach(consumer);
      }
    }

    @Override
    public void forEachOrdered(DoubleConsumer consumer) {
      if (!isParallel()) {
        adapt(sourceStageSpliterator()).forEachRemaining(consumer);
      } else {
        super.forEachOrdered(consumer);
      }
    }

  }

  /**
   * Base class for a stateless intermediate stage of a DoubleStream.
   *
   * @param <E_IN> type of elements in the upstream source
   * @since 1.8
   */
  abstract static class StatelessOp<E_IN> extends DoublePipeline<E_IN> {

    /**
     * Construct a new DoubleStream by appending a stateless intermediate
     * operation to an existing stream.
     *
     * @param upstream the upstream pipeline stage
     * @param inputShape the stream shape for the upstream pipeline stage
     * @param opFlags operation flags for the new stage
     */
    StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
        StreamShape inputShape,
        int opFlags) {
      super(upstream, opFlags);
      assert upstream.getOutputShape() == inputShape;
    }

    @Override
    final boolean opIsStateful() {
      return false;
    }
  }

  /**
   * Base class for a stateful intermediate stage of a DoubleStream.
   *
   * @param <E_IN> type of elements in the upstream source
   * @since 1.8
   */
  abstract static class StatefulOp<E_IN> extends DoublePipeline<E_IN> {

    /**
     * Construct a new DoubleStream by appending a stateful intermediate
     * operation to an existing stream.
     *
     * @param upstream the upstream pipeline stage
     * @param inputShape the stream shape for the upstream pipeline stage
     * @param opFlags operation flags for the new stage
     */
    StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
        StreamShape inputShape,
        int opFlags) {
      super(upstream, opFlags);
      assert upstream.getOutputShape() == inputShape;
    }

    @Override
    final boolean opIsStateful() {
      return true;
    }

    @Override
    abstract <P_IN> Node<Double> opEvaluateParallel(PipelineHelper<Double> helper,
        Spliterator<P_IN> spliterator,
        IntFunction<Double[]> generator);
  }
}
